package net.hn.hnms.biz.upkafka.service;

import net.hn.hnms.biz.upkafka.config.KafkaProperties;
import net.hn.hnms.biz.upkafka.config.SpringContextUtil;
import net.hn.hnms.biz.upkafka.domain.FanrealNowPo;
import net.hn.hnms.biz.upkafka.domain.InclineHoistingNowPo;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * 斜井提升
 */
public class XJProducer extends Thread {
    private static final Logger LOG = LoggerFactory.getLogger(Producer.class);

    public final KafkaProducer<String, String> producer;

    private final String topic;
    private final String mineCode;

    public final Boolean isAsync;
    public XJProducer(String topicName, Boolean asyncEnable, String mineCode) {
        Properties props = Producer.initProperties(topicName,mineCode);
        producer = new KafkaProducer<String, String>(props);
        topic = topicName;
        isAsync = asyncEnable;
        this.mineCode = mineCode;
    }
    @Override
    public void run() {
        // 构造消息记录
        while (true) {
            StringBuffer message = this.getXJSSMessage(this.mineCode);
            long startTime = System.currentTimeMillis();
            String key = String.valueOf(mineCode);
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, key, message.toString());
            if (isAsync) {
                // 异步发送
                producer.send(record);
            } else {
                try {
                    // 同步发送
                    producer.send(record).get();
                    System.out.println("发送成功:"+message);
                } catch (InterruptedException ie) {
                    LOG.info("The InterruptedException occured : {}.", ie);
                } catch (ExecutionException ee) {
                    LOG.info("The ExecutionException occured : {}.", ee);
                }
            }
            try {
                // 线程阻塞 1 分钟 (60000 毫秒)
                Thread.sleep(60000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
    private StringBuffer getXJSSMessage(String mineCode) {
        String mineName = KafkaProperties.getMineNameByMineCode(mineCode);
        String systemName = "斜井提升系统";
        String systemCode = "XJSS";
        String statusCode="33007";  //测点类型编码-运行状态
        String Kg = "KG";
        String deviceCode1 = "000000000001";
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        //数据上传时间
        LocalDateTime now = LocalDateTime.now();
        // Define the desired format
        DateTimeFormatter uploadedTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        // Format the current date and time
        //数据上传时间
        String uploadedTime = now.format(uploadedTimeFormatter);
        //构建消息内容
        StringBuffer message = new StringBuffer();
        //主通风系统接入频率：每60s上传一次测点实时数据。
        //构建消息头：包括煤矿编码，煤矿名称，系统型号，系统名称，数据上传时间
        String messageHeader = mineCode + ";"+mineName+";;"+systemName+";" + simpleDateFormat.format(new Date()) + "~";
        message.append(messageHeader);

        StringBuffer messageBody = new StringBuffer();
        //构建主通风消息体：测点编码，测点类型编码，测点类型名称，所属设备编码，测点数值，测点数值单位，测点状态，数据时间~

        //获取 斜井实时数据
        SystemService systemService = SpringContextUtil.getBean(SystemService.class);
        //获取数据
        Map<String, InclineHoistingNowPo> InclineHoistingNowMap = systemService.getXJSSSystemStatus(mineCode);
        //数据不为空
        if(InclineHoistingNowMap!=null && InclineHoistingNowMap.size()>0){
            String xjStart1="";
            if ("150781013671".equals(mineCode)) {
                //灵露
                xjStart1 = "NewSite_副井绞车_提升机运行";
            }else if ("620824009734".equals(mineCode)) {
                //研北
                xjStart1 = "";
            }else if ("150621011015".equals(mineCode)) {
                //高头窑
                xjStart1 = "";
            }

            //系统状态
            InclineHoistingNowPo po1 = InclineHoistingNowMap.get(xjStart1);
            if (po1 != null) {
                FanrealNowPo vo1 = new FanrealNowPo();
                vo1.setDataTime(po1.getDataTime() != null ? po1.getDataTime() : new Date());
                vo1.setPointValue(po1.getPointValue() != null ? po1.getPointValue() : 0);
                vo1.setPointStatus(po1.getPointStatus() != null ? po1.getPointStatus() : 1);
                messageBody.append(mineCode+systemCode+statusCode+Kg+deviceCode1+";"+statusCode+";运行状态;"+systemCode+";" + vo1.getPointValue().intValue()+ ";;" + vo1.getPointStatus() + ";" +simpleDateFormat.format(vo1.getDataTime()) + "~");
            }
        }


        //构建消息结尾
        messageBody.append("||");
        message.append(messageBody);
        return message;
    }
}
